package net.qiqbframework.modelling.command;


import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import net.qiqbframework.commandhandling.*;
import net.qiqbframework.common.Registration;
import net.qiqbframework.common.lock.LockFactory;
import net.qiqbframework.common.lock.PessimisticLockFactory;
import net.qiqbframework.common.lock.TopicLock;
import net.qiqbframework.common.utils.CollectionUtil;
import net.qiqbframework.common.utils.ReflectionUtil;
import net.qiqbframework.config.ConfigurationException;
import net.qiqbframework.messaging.annotation.MessageExecutableHandler;
import net.qiqbframework.messaging.task.BatchingMessageTask;
import net.qiqbframework.messaging.task.CurrentMessageTask;
import net.qiqbframework.messaging.task.MessageTask;
import net.qiqbframework.modelling.command.voucher.CommandLoadVoucher;
import net.qiqbframework.modelling.domain.Aggregate;
import net.qiqbframework.modelling.domain.AggregateModel;
import net.qiqbframework.modelling.domain.AggregateNotFoundException;
import net.qiqbframework.modelling.domain.BizIdentifierModel;
import net.qiqbframework.modelling.repository.AggregateBizIdentifierVoucher;
import net.qiqbframework.modelling.repository.AggregateRepository;
import net.qiqbframework.serialization.SerializedObject;
import net.qiqbframework.serialization.Serializer;

import java.lang.reflect.Executable;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

@Slf4j
public class AggregateAnnotationCommandHandler<T> implements CommandMessageHandler {

    private final AggregateRepository<T> repository;

    private final CommandTargetResolver commandTargetResolver;

    private final Map<String, Set<CommandMessageHandler>> supportedCommandsByName;

    @Getter
    private final AggregateModel<T> aggregateModel;


    private final AggregateHandlingMonitor aggregateHandlingMonitor;

    private final Serializer aggregateRootSerializer;
    // private LockFactory lockFactory = PessimisticLockFactory.usingDefaults();

    protected AggregateAnnotationCommandHandler(AggregateAnnotationCommandHandlerBuilder<T> builder) {

        builder.validate();

        this.repository = builder.repository;
        this.aggregateHandlingMonitor = builder.aggregateHandlingMonitor();
        this.aggregateRootSerializer = builder.aggregateRootSerializer();
        this.commandTargetResolver = builder.commandTargetResolver;
        this.supportedCommandsByName = new HashMap<>();
        AggregateModel<T> aggregateModel = builder.buildAggregateModel();
        this.aggregateModel = aggregateModel;
        initializeHandlers(aggregateModel);

    }

    public Registration subscribe(CommandBus commandBus) {
        List<Registration> subscriptions = supportedCommandsByName.entrySet().stream()
                // flatMap
                .flatMap(entry -> entry.getValue().stream().map(messageHandler -> commandBus.subscribe(entry.getKey(), this)))
                // filter
                .filter(Objects::nonNull).toList();
        return () -> subscriptions.stream().map(Registration::cancel).reduce(Boolean::logicalOr).orElse(false);
    }

    private void initializeHandlers(AggregateModel<T> aggregateModel) {

        aggregateModel.allCommandHandlers().values().stream().flatMap(List::stream)
                // collect
                .collect(Collectors.groupingBy(this::getHandlerSignature))
                //
                .forEach((signature, commandHandlers) -> initializeHandler(aggregateModel, commandHandlers.get(0)));


    }

    private String getHandlerSignature(MessageExecutableHandler handler) {
        return handler.unwrap(Executable.class).map(ReflectionUtil::toDiscernibleSignature).orElseThrow(() -> new IllegalStateException("A handler is missing an Executable. Please provide an " + "Executable in your MessageHandlingMembers"));
    }

    private void initializeHandler(AggregateModel<T> aggregateModel, MessageExecutableHandler handler) {

        handler.unwrap(CommandMessageExecutableHandler.class).ifPresent(cmh -> {
            CommandMessageHandler messageHandler;
            if (cmh.isFactoryHandler()) {
                messageHandler = new AggregateConstructorCommandHandler(handler);
            } else {
                messageHandler = new AggregateCommandHandler(cmh);
            }
            supportedCommandsByName.computeIfAbsent(cmh.commandNaming(), key -> new HashSet<>()).add(messageHandler);
        });
    }

    @Override
    public Object handle(CommandMessage<?> commandMessage) throws Exception {
        final Set<CommandMessageHandler> commandMessageHandlers = supportedCommandsByName.get(commandMessage.getCommandName());
        if (commandMessageHandlers == null || commandMessageHandlers.isEmpty()) {
            throw new CommandHandlerNotFoundException(commandMessage);
        }
        return commandMessageHandlers.stream().findFirst().get().handle(commandMessage);
    }


    protected Object resolveReturnValue(@SuppressWarnings("unused") CommandMessage<?> command, Aggregate<T> createdAggregate) {
        return createdAggregate.identifier();
    }


    private class AggregateConstructorCommandHandler implements CommandMessageHandler {

        private final MessageExecutableHandler handler;

        public AggregateConstructorCommandHandler(MessageExecutableHandler handler) {
            this.handler = handler;
        }

        @SuppressWarnings("unchecked")
        @Override
        public Object handle(CommandMessage<?> command) throws Exception {
            aggregateHandlingMonitor.beforeHandle(command, null);
            Aggregate<T> aggregate = repository.newInstance(() -> (T) handler.handle(command, null));

            final SerializedObject<byte[]> afterHandleSerialize = aggregateRootSerializer.serialize(aggregate.aggregateRoot(), byte[].class);
            final T afterHandlSnapshot = aggregateRootSerializer.deserialize(afterHandleSerialize);
            // 添加快照
            CurrentMessageTask.get().afterCommit((t) -> {
                aggregateHandlingMonitor.afterCommit(command, null, afterHandlSnapshot, aggregate);
            });
            return resolveReturnValue(command, aggregate);
        }
    }

    private class AggregateCommandHandler implements CommandMessageHandler {

        private final CommandMessageExecutableHandler handler;

        public AggregateCommandHandler(CommandMessageExecutableHandler handler) {
            this.handler = handler;
        }


        @Override
        public Object handle(CommandMessage<?> command) throws Exception {
            //
            // 尝试从缓存中加载，如果有直接拿
            final List<CommandLoadVoucher> commandLoadVouchers = splitGroupAggregate(command);

            // 初步加锁，
            for (CommandLoadVoucher bizIdentifierVoucher : commandLoadVouchers) {
                if (bizIdentifierVoucher.getPrimaryBizIdentifierMember().isEmpty()) {
                    continue;
                }
                final BizIdentifierVoucherMember primaryBizIdentifierMember = bizIdentifierVoucher.getPrimaryBizIdentifierMember().get();
                final String topicName = aggregateModel.modelName() + "#" + primaryBizIdentifierMember.getBizIdentifierName() + ":" + primaryBizIdentifierMember.getValue();
                //TopicLock lock = lockFactory.obtainLock(topicName);
                //CurrentMessageTask.get().onCleanup(u -> lock.release());
            }
            // 当前命令的加载信息
            final CommandLoadVoucher currCommandLoadVoucher = commandLoadVouchers.get(0);
            // 聚合对象
            Aggregate<T> aggregate = gettAggregate(command, currCommandLoadVoucher, commandLoadVouchers);

            // 使用ID加锁
            /*TopicLock lock = lockFactory.getLock(aggregate.globalIdentifierAsString());
            if (!lock.tryLock()) {
                // todo 这里可以优化。强制刷新，重新加载。
                throw new IllegalStateException("当前聚合业务正在操作，请稍后重试");
            }
            CurrentMessageTask.get().onCleanup(u -> lock.release());*/
            // 如何判断当前加载的聚合根对象是最新的。如果不是最新的，将重新在加载一遍
            // todo 需要验证所有加载凭证是否匹配，如果不匹配，将报错
            final Map<BizIdentifierModel<?>, Object> bizIdentifiers = aggregateModel.getBizIdentifiers(aggregate.aggregateRoot());

            for (BizIdentifierVoucherMember subBizIdentifierMember : commandLoadVouchers.get(0).listSubBizIdentifierMembers()) {
                final Object commandSubBizId = subBizIdentifierMember.getValue();
                final String commandSubBizIdName = subBizIdentifierMember.getBizIdentifierName();
                final Object aggregateBizValue = bizIdentifiers.get(aggregateModel.getBizIdentifierModel(commandSubBizIdName));
                if (!commandSubBizId.equals(aggregateBizValue)) {
                    // 如果存在子业务ID，肯定存在主业务ID信息
                    final BizIdentifierVoucherMember currBizIdentifierVoucherMember = currCommandLoadVoucher.getPrimaryBizIdentifierMember().orElseThrow();
                    throw new MultiBizIdentifierDisagreementException("The aggregate [%s] loaded by [%s] does not match the [%s] in the command.".formatted(commandSubBizIdName + ":" + aggregateBizValue,
                            currBizIdentifierVoucherMember.getValue(), commandSubBizIdName + ":" + commandSubBizId));
                }
            }
            // 进行虚拟化
            final SerializedObject<byte[]> beforeHandleSerialize = aggregateRootSerializer.serialize(aggregate.aggregateRoot(), byte[].class);
            final T beforeHandleSnapshot = aggregateRootSerializer.deserialize(beforeHandleSerialize);
            aggregateHandlingMonitor.beforeHandle(command, beforeHandleSnapshot);
            final Object result = aggregate.handle(command);
            final SerializedObject<byte[]> afterHandleSerialize = aggregateRootSerializer.serialize(aggregate.aggregateRoot(), byte[].class);
            final T afterHandlSnapshot = aggregateRootSerializer.deserialize(afterHandleSerialize);
            // 添加快照
            Aggregate<T> finalAggregate = aggregate;
            CurrentMessageTask.get().afterCommit((t) -> {
                aggregateHandlingMonitor.afterCommit(command, beforeHandleSnapshot, afterHandlSnapshot, finalAggregate);
            });
            return result;
        }

        private Aggregate<T> gettAggregate(CommandMessage<?> command, CommandLoadVoucher currCommandLoadVoucher, List<CommandLoadVoucher> commandLoadVouchers) {
            Aggregate<T> aggregate;
            if (currCommandLoadVoucher.getPrimaryBizIdentifierMember().isPresent()) {
                final BizIdentifierVoucherMember currBizIdentifierVoucherMember = currCommandLoadVoucher.getPrimaryBizIdentifierMember().get();
                final AggregateBizIdentifierVoucher currBizIdentifierVoucher = toAggregateBizIdentifierVoucher(currBizIdentifierVoucherMember);
                // 尝试从缓存中获取
                aggregate = repository.loadFromCache(currBizIdentifierVoucher);
                if (aggregate == null) {
                    if (commandLoadVouchers.size() > 1) {
                        // 存在批量，需要执行预备加载，提高性能
                        final List<AggregateBizIdentifierVoucher> aggregateBizIdentifierVouchers = new ArrayList<>(commandLoadVouchers.subList(1, commandLoadVouchers.size()).stream()
                                // 转化
                                .map(v -> toAggregateBizIdentifierVoucher(v.getPrimaryBizIdentifierMember().orElse(v.getFirstBizIdentifierMember())))
                                .toList());
                        // 当前加载信息的添加到第一个
                        aggregateBizIdentifierVouchers.add(0, currBizIdentifierVoucher);
                        repository.prepareBatchLoad(aggregateBizIdentifierVouchers);
                    }
                }
                aggregate = repository.load(currBizIdentifierVoucher);
                if (aggregate == null) {
                    throw new AggregateNotFoundException(currBizIdentifierVoucher.getBizIdentifier().toString(), "The loading info [%s] for command [%s#%s] did not find the corresponding aggregation root.".formatted(currBizIdentifierVoucher.getBizIdentifier().toString(),
                            command.getPayload().getClass(), currBizIdentifierVoucherMember.getMember().getName()));
                }
                Aggregate<T> finalAggregate = aggregate;
                CurrentMessageTask.get().getOrComputeResource("_lastAggregate", (Function<String, Object>) s -> finalAggregate);
            } else {
                // 没有业务ID，查看上一个聚合根是否一致。如果类型一致，直接返回
                final Object _lastAggregate = CurrentMessageTask.get().getResource("_lastAggregate");
                if (_lastAggregate instanceof Aggregate lastAggregate && lastAggregate.aggregateName().equals(aggregateModel.modelName())) {
                    aggregate = lastAggregate;

                } else {
                    throw new AggregateNotFoundException("", "No loading information was found in the  [%s] command .".formatted(command.getPayload().getClass()));
                }
            }
            return aggregate;
        }

        /**
         * 探测之后的命令和当前的命令操作同一种聚合对象，
         *
         * @param indexCommand 当前命令
         * @return
         */
        private List<CommandLoadVoucher> splitGroupAggregate(CommandMessage<?> indexCommand) {
            List<CommandLoadVoucher> result = new ArrayList<>();
            final MessageTask<?> messageTask = CurrentMessageTask.get();
            String notFoundErrorMessage = "The command [%s] could not find the aggregation bizId for loading. Please use the @BizIdentifierVoucher.";
            if (messageTask instanceof BatchingMessageTask<?> batchingMessageTask) {
                for (Object message : CollectionUtil.subRemainingByTarget(batchingMessageTask.getMessages(), indexCommand)) {
                    final CommandLoadVoucher commandLoadVoucher = commandTargetResolver.resolveLoadVoucher((CommandMessage<?>) message);
                    if (!commandLoadVoucher.hasBizIdentifierMember()) {
                        throw new CommandConfigurationException(notFoundErrorMessage.formatted(((CommandMessage<?>) message).getCommandName()));
                    }
                    result.add(commandLoadVoucher);
                }
            } else {
                // 仅仅解析当前的
                final CommandLoadVoucher commandLoadVoucher = commandTargetResolver.resolveLoadVoucher(indexCommand);
                if (commandLoadVoucher.getPrimaryBizIdentifierMember().isEmpty()) {
                    throw new CommandConfigurationException(notFoundErrorMessage.formatted(indexCommand.getCommandName()));
                }
                result.add(commandLoadVoucher);
            }

            return result;
        }

        private AggregateBizIdentifierVoucher toAggregateBizIdentifierVoucher(BizIdentifierVoucherMember bizIdentifierMember) {

            String businessIdName = bizIdentifierMember.getBizIdentifierName();
            final BizIdentifierModel<?> businessIdModel = aggregateModel.getBizIdentifierModel(businessIdName);
            if (businessIdModel == null) {
                throw new ConfigurationException("The bizId [%s] is not defined in the aggregation class [%s].".formatted(businessIdName, aggregateModel.entityClass()));
            }
            return new AggregateBizIdentifierVoucher(businessIdName, bizIdentifierMember.getValue());
        }
    }

}
